package com.yixiu.bi.hbase;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * HBase操作类
 * 连接池
 */
public class HBaseClient {
	private static ConcurrentHashMap<String,HBaseClient> hbaseMap = new ConcurrentHashMap<String, HBaseClient>();

	private  Logger logger = LoggerFactory.getLogger(this.getClass());

	private String configFile = ""; // 配置文件
	private  Configuration configuration;
	private  Connection connection;// 这是一个线程安全的连接池，一个程序共用一个连接池，程序结束时关闭连接池就行了
	private  Properties properties;

	/**
	 * 根据配置文件初始化HBaseClient
	 * @param configFile 配置文件
	 */
	private HBaseClient(String configFile) {
		this.configFile = configFile;
		// 读取配置，初始化producer 和 topic

		try {
			properties = new Properties();
			properties.load(HBaseClient.class.getResourceAsStream(configFile));

			// 读取 hbase.zookeeper.quorum
			String zookeeperIp = properties.getProperty("hbase.zookeeper.quorum");
			if(StringUtils.isEmpty(zookeeperIp)) {
				logger.error( configFile + "没有配置hbase.zookeeper.quorum");
				return;
			}

			// 读取 hbase.zookeeper.property.clientPort
			String zookeeperPort = properties.getProperty("hbase.zookeeper.property.clientPort");
			if(StringUtils.isEmpty(zookeeperPort)) {
				logger.error( configFile + "没有配置hbase.zookeeper.property.clientPort");
				return;
			}

			configuration = HBaseConfiguration.create();
			configuration.set("hbase.zookeeper.quorum", zookeeperIp);
			configuration.set("hbase.zookeeper.property.clientPort", zookeeperPort);
			configuration.set("hbase.defaults.for.version.skip", "true");
			connection = ConnectionFactory.createConnection(configuration);

		}catch (IOException ex) {
			logger.error("配置文件不存在:" + configFile,ex);
		} catch (Exception ex) {
			logger.error("创建HBaseClient异常:" + configFile,ex);
			this.connection = null;
		}
	}


	/**
	 * 根据配置文件，获取HBase客户端
	 * @param configFile 配置文件
	 * @return
	 */
	public static HBaseClient getHBaseClient(String configFile) {
		if(hbaseMap.containsKey(configFile)) {
			return hbaseMap.get(configFile);
		}

		HBaseClient  hbaseClient  = new HBaseClient(configFile);
		if(hbaseClient.connection != null) {
			synchronized (hbaseMap) {
				if(!hbaseMap.containsKey(configFile)) {
					hbaseMap.put(configFile,hbaseClient);
				}
			}
		}
		return  hbaseClient;
	}




	/**
	 * 插入一行数据，可以指定多列
	 * @param tableName 表名称
	 * @param rowkey 行主键 rowkey
	 * @param values 要插入的列值，数据的每一行是一列数据，values[i][0]=family,values[i][1]=qualifier,values[i][2]=value
	 * @return 插入成功返回true，插入失败返回false
	 */
	public boolean insertData(String tableName, String rowkey, String[][] values) {

		boolean result = false;
		if (values == null || values.length == 0) return result;
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(rowkey.getBytes());// 一个PUT代表一行数据，再NEW一个PUT表示第二行数据,每行一个唯一的ROWKEY，此处rowkey为put构造方法中传入的值
			for (String[] strings : values) {
				put.addColumn(strings[0].getBytes(), strings[1].getBytes(), strings[2].getBytes());
			}
			table.put(put);
			result = true;
		} catch (Exception e) {
			logger.error("HbaseClient insertData exception", e);
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (Exception e) {
					logger.error("HbaseClient table.close exception:", e);
				}
			}
		}
		return result;
	}


	/**
	 * 一次插入多行数据
	 * @param tableName 表名
	 * @param map key是行关键字rowkey,value是多列数据，values[i][0]=family,values[i][1]=qualifier,values[i][2]=value
	 * @return 插入成功返回true,失败返回false
	 */
	public boolean insertDatas(String tableName,Map<String,String[][]> map){

		boolean result = false;
		if (map == null || map.size() == 0) return result;
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			List<Put> list = new ArrayList<Put>();
			for (String rowkey: map.keySet() ) {
				Put put = new Put(rowkey.getBytes());
				String [][] cols = map.get(rowkey); // 1行对应的多列
				for(int i = 0; i < cols.length; i++) {
					put.addColumn(cols[i][0].getBytes(), cols[i][1].getBytes(), cols[i][2].getBytes());
				}
				list.add(put);
			}
			table.put(list);

			result = true;
		} catch (Exception e) {
			logger.error("HbaseClient insertData exception", e);
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (Exception e) {
					logger.error("HbaseClient table.close exception:", e);
				}
			}
		}
		return result;
	}


	/**
	 * 插入一行数据，可以指定多列，插入前可以判断某一列的值是否为预期值
	 * HBase通过CAS(compare-and-set)操作来解决数据一致性问题
	 * 只有当表中expectedFamily:expectedQualifier对应的值等于expectedValue时才插入
	 * @param tableName 表名称
	 * @param rowkey rowkey
	 * @param expectedFamily 要检查的列簇
	 * @param expectedQualifier 要检查的列
	 * @param expectedValue 要检查的值
	 * @param values 数据的每一行是一列数据，values[i][0]=family,values[i][1]=qualifier,values[i][2]=value
	 * @return 插入成功返回true,失败返回false
	 */
	public boolean checkAndPut(String tableName, String rowkey, String expectedFamily, String expectedQualifier, String expectedValue,
			String[][] values) {
		boolean result = false;
		if (values == null || values.length == 0) return result;

		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Put put = new Put(Bytes.toBytes(rowkey));

			for (String[] strings : values) {
				put.addColumn(Bytes.toBytes(strings[0]), Bytes.toBytes(strings[1]), Bytes.toBytes(strings[2]));
			}
			if (expectedValue == null) {
				result = table.checkAndPut(Bytes.toBytes(rowkey), Bytes.toBytes(expectedFamily), Bytes.toBytes(expectedQualifier),
							CompareOp.EQUAL, null, put);
			} else {
				result = table.checkAndPut(Bytes.toBytes(rowkey), Bytes.toBytes(expectedFamily), Bytes.toBytes(expectedQualifier),
						CompareOp.EQUAL, Bytes.toBytes(expectedValue), put);
			}
		} catch (Exception e) {
			logger.error("HbaseClient exception", e);

		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (Exception e) {
					logger.error("HbaseClient table.close exception:", e);
				}
			}
		}
		return result;
	}
	




	/**
	 * 查询一行数据的某列的值
	 * @param tableName 表名
	 * @param rowkey 行主键
	 * @param family 列簇
	 * @param column 列名
	 * @return 返回列值
	 */
	public String queryByColumn(String tableName, String rowkey, String  family, String column) {
    	String [] fams = { family };
    	String [] cols = { column };
		Map<String,String> map = queryByColumn(tableName,rowkey, fams,cols);
		String value = "";
		if(map != null && map.size() > 0) {
			for (String key: map.keySet() ) {
				value = map.get(key);
				break;
			}
		}
		return  value;
	}

	/**
	 * 查询一行数据对应的列值
	 * @param tableName 表名
	 * @param rowkey 行主键 rowkey
	 * @param family 多列对应的family
	 * @param columns 多个列名
	 * @return 返回列名和列值的键值对
	 */
	public Map<String, String> queryByColumn(String tableName, String rowkey, String [] family, String[] columns) {
    	if(family == null || columns == null || family.length != columns.length || columns.length == 0) {
    		return  null;
		}
		Map<String, String> map = new HashMap<String,String>();
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Get scan = new Get(rowkey.getBytes());
			for (int i = 0; i < columns.length; i++) {
				scan.addColumn(family[i].getBytes(), columns[i].getBytes());
			}
			Result r = table.get(scan);
			if (CollectionUtils.isNotEmpty(r.listCells())) {
				for (Cell cell : r.listCells()) {
					map.put(Bytes.toString(CellUtil.cloneQualifier(cell)),Bytes.toString( CellUtil.cloneValue(cell)));
				}
			}
		} catch (Exception e) {
			logger.error("HbaseClient exception", e);

		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (Exception e) {
					logger.error("HbaseClient table.close exception:", e);
				}
			}
		}
		return map;
	}


	/**
	 * hbase单计数器
	 *  针对某一列进行计数
	 * @param tableName
	 * @param row
	 * @param family
	 * @param qualifier
	 * @param amount
	 * @return
	 */
	public Long incrementColumnValue(String tableName, String row, String family, String qualifier, final long amount) {
		Table table = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));;
			return table.incrementColumnValue(Bytes.toBytes(row), Bytes.toBytes(family), Bytes.toBytes(qualifier), amount);
		} catch (IOException e) {

			logger.error("exception:"+e.getStackTrace());
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (Exception e) {
					logger.error("HbaseClient table.close exception:", e);

				}
			}
		}
		return null;
	}

	/**
	 * hbase多计数器，注意qualifierArray与amountArray数据要一一对应
	 *
	 * @param tableName  表名
	 * @param row rowkey
	 * @param family 列簇
	 * @param qualifierArray 列名
	 * @param amountArray  列值(数值)
	 * @return
	 */
	public long[] increment(String tableName, String row, String family, String[] qualifierArray, final long[] amountArray) {
		Table table = null;
		long[] values = null;
		try {
			table = connection.getTable(TableName.valueOf(tableName));
			Increment increment = new Increment(Bytes.toBytes(row));
			for (int i=0; i<qualifierArray.length; i++) {
				increment.addColumn(family.getBytes(), qualifierArray[i].getBytes(), amountArray[i]);
			}
			Result result = table.increment(increment);
			values = new long[qualifierArray.length];
			for (int i=0; i<qualifierArray.length; i++) {
				values[i] = Bytes.toLong(result.getValue(family.getBytes(), qualifierArray[i].getBytes()));
			}
		} catch (IOException e) {
			logger.error("exception:"+e.getStackTrace());

		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (Exception e) {
					logger.error("HbaseClient table.close exception:", e);

				}
			}
		}
		return values;
	}



}
